<?php
namespace zake\foundation\reactor\trigger;

use zake\helper\Console;
use zake\helper\Grant;
use zake\helper\Session;
use zake\foundation\able\BeeAble;
use zake\foundation\stream\ReadStream;
use zake\foundation\able\traits\PointTrait;
use zake\foundation\able\reactor\ChannelAble;
use zake\foundation\able\reactor\ConnectAble;
use zake\foundation\able\reactor\ListenAble;
use zake\foundation\reactor\trigger\connect\Connect;


class Listen extends ReadStream implements ListenAble
{
    use PointTrait;

    /**
     * connect class name
     *
     * @var string
     */
    private $connectClass = Connect::class;


    /**
     * 与主进程通信管道
     * @var null | ChannelAble
     */
    private $channel = null;


    /**
     * set connect class
     * @param $className
     *
     * @author liu.bin 2020-05-06 22:12
     */
    public function setConnect($className)
    {
        $this->connectClass = $className;
    }


    /**
     * 获取connect
     *
     * @return null|ConnectAble
     *
     * @author liu.bin 2020-04-28 10:26
     */
    public function acceptConnect()
    {

        // 惊群的问题
        // 多进程模式下，listenSocket可读时，所有进程会被唤醒，只有一个进程会accept成功
        // 如果返回null，表明该进程被唤醒，但是没有accept成功，需要再次进入reactor监听
        $socket = @stream_socket_accept($this->getResource(),0);
        if(!is_resource($socket)){
            return null;
        }

        $connectId = Grant::permitConnectId();
        if(!$connectId){
            fclose($socket);
            Console::error("accept failed. the accept queue is full. left size is [".Grant::connectSize()."]");
            return null;
        }

        /**
         * @var BeeAble|ConnectAble $connect
         */
        stream_set_blocking($socket,0);
        $className  = $this->connectClass;
        $connect    = new $className($socket);
        //
        $connect->setListener($this);
        $connect->setProtocol($this->getProtocol());
        $connect->setConnectId($connectId);

        // todo session 相关
        //$processId = 1;
        //$clusterId = 1;
        //$connect->setSessionId(Session::id($clusterId,$processId,$connectId));
        return $connect;
    }


    /**
     * 构造器
     *
     * Channel constructor.
     */
    protected function __construct($uri)
    {
        parent::__construct($this->createListenSocket($uri));
    }


    /**
     * 定义与主进程通信channel
     * @param ChannelAble $channel
     *
     * @author liu.bin 2020-05-15 16:30
     */
    public function setChannel(ChannelAble $channel)
    {
        $this->channel = $channel;
    }

    /**
     * 单机进程通信管道
     * @return ChannelAble|BeeAble
     *
     * @author liu.bin 2020-05-15 23:43
     */
    public function getChannel()
    {
        return $this->channel;
    }



    /**
     * 初始化监听socket
     *
     * @author liu.bin 2020-04-15 14:20
     */
    private static function createListenSocket($uri)
    {
        if(is_string($uri)){
            $serverSocket = @stream_socket_server(
                $uri,
                $errNo, $errStr,
                STREAM_SERVER_BIND | STREAM_SERVER_LISTEN,
                self::getContextResource()
            );
            if(!$serverSocket){
                Console::error("create server fail");
                exit;
            }
            $socket = socket_import_stream($serverSocket);
            @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
            @socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
            stream_set_blocking($serverSocket, 0);

        }elseif (is_resource($uri)){
            $serverSocket = $uri;
        }

        return $serverSocket;
    }



    /**
     *
     * @return resource
     * @author liu.bin 2020-04-12 00:42
     */
    private static function getContextResource()
    {
        $opt = [];
        $opt['socket'] = ['backlog' => 0];
        $contextResource = stream_context_create($opt);
        stream_context_set_option($contextResource, 'socket', 'so_reuseport', 1);
        return $contextResource;
    }


}
